package org.example.config.mqtt;

import org.example.config.mqtt.property.MqttProperty;
import org.example.device.mqtt.event.DeviceNotifyMqttEvent;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.ExecutorChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
import org.springframework.integration.transformer.GenericTransformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@ConditionalOnBean(MqttConfig.class)
public class MqttProducerConfig {
    @Bean
    public MessageChannel mqttReplyChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageChannel mqttOutboundExecutorChannel() {
        return new ExecutorChannel(asyncExecutor());
    }

    @Bean
    public ThreadPoolTaskExecutor asyncExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("Async-Gateway-");
        executor.initialize();
        return executor;
    }
    @Bean
    public MqttPahoMessageHandler mqttOutbound(DefaultMqttPahoClientFactory mqttClientFactory, MqttProperty mqttProperty, MessageConverter messageConverter) {
        MqttPahoMessageHandler handler = new MqttPahoMessageHandler(
                mqttProperty.getClientId()+"-producer",//自动生成客户端Id
                mqttClientFactory);
        handler.setAsync(true); // 异步发送
        handler.setDefaultTopic("test/topic"); // 默认主题
        handler.setDefaultQos(0);// 默认 QoS 级别
        handler.setConverter(messageConverter);
        return handler;
    }

    @Bean
    public IntegrationFlow mqttOutboundFlow(DefaultMqttPahoClientFactory mqttClientFactory, MessageConverter messageConverter, MqttProperty mqttProperty) {
        return IntegrationFlows.from(mqttOutboundExecutorChannel())
                .handle(mqttOutbound(mqttClientFactory, mqttProperty, messageConverter))
                .get();
    }

    @Bean
    public MessageChannel producerErrorChannel() {
        return new ExecutorChannel(producerTaskErrorExecutor());
    }

    @Bean
    public ThreadPoolTaskExecutor producerTaskErrorExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.setThreadNamePrefix("MQTT-Error-Produce-");
        executor.initialize();
        return executor;
    }
}
